Conversation
|
Thanks @peterxcli please check on your fork https://github.com/apache/datafusion-comet/actions/workflows/spark_sql_writer_tests.yml |
|
@comphead would you like to take a look? Thanks! |
|
Thanks @peterxcli there is another angle #5293 |
…Spark 4.0+ Native writes replace the whole DataWritingCommandExec, which means InsertIntoHadoopFsRelationCommand.run never runs. Everything that method does has to be re-implemented inside CometNativeWriteExec: a hardcoded SQLHadoopMapReduceCommitProtocol (so spark.sql.sources.commitProtocolClass is ignored), dynamicPartitionOverwrite pinned to false, a hand-ported copy of the SaveMode logic, a bespoke commit-message accumulator, and its own commitJob call. Most of the open native-writer issues are symptoms of that one decision rather than independent defects. On Spark 4.0+, V1WritesUtils.getWriteFilesOpt matches the WriteFilesExecBase trait (introduced in 4.0 precisely for this), so a Comet node that extends it gets driven through FileFormatWriter.executeWrite -> SparkPlan.executeWrite -> doExecuteWrite, and Spark keeps ownership of everything above the per-task write. Spark 3.x has no such trait: getWriteFilesOpt matches the concrete WriteFilesExec case class, a Comet node there would not be found, and Spark would silently take FileFormatWriter's non-planned, row-based branch. So the new seam is additive. CometDataWritingCommand and CometNativeWriteExec are kept unchanged and remain the 3.4/3.5 path; CometExecRule picks the path by version and the two never both fire. The legacy path goes away with Spark 3.x support. Add: - CometWriteFilesExec, overriding doExecuteWrite and mirroring FileFormatWriter.executeTask for the parts Comet must do itself: build the TaskAttemptContext, ask the commit protocol for a path, run the native writer, drive the stats trackers, commit or abort. Plus the CometWriteFiles serde and a two-line ShimCometWriteFilesExec in spark-4.x / spark-3.x. - File paths come from FileCommitProtocol.newTaskTempFile and are used verbatim, so names match Spark's part-<id>-<uuid>-c000.<codec>.parquet and committers that track individual files (S3A magic, streaming manifest) work. - Column names, nullability and field IDs come from WriteJobDescription.dataColumns rather than the query output, so INSERT INTO t SELECT a+1 writes the target column's name. - Byte and row counts come from BasicWriteTaskStatsTracker, which stats files through the FileSystem API and is therefore correct on HDFS. - ParquetWriter proto: work_dir is now optional. When set (3.x) the native writer derives the file name as before; when unset (4.0+) output_path is the exact file to write and is used verbatim. - On 4.0+ the opt-in moves to spark.comet.operator.WriteFilesExec .allowIncompatible, with the old DataWritingCommandExec key kept as a deprecated alternative. isOperatorAllowIncompat now resolves alternatives, which the planner's by-name lookup previously bypassed. AQE re-plans the write command's child and re-inserts a WriteFilesExec above the node Comet already converted; leaving DataWritingCommandExec in place means Comet no longer has to guard against the resulting nested native writes.
…configs Matches the reviewed form on apache#5293: ConfigBuilder mutates in place, so the Seq destructuring was rebinding the same object. Only one operator has an alternative and there is no reason to expect more.
The output path reaches both write serdes as `Path.toString`, which decodes percent escapes: a directory containing a space or a literal `%` yields a string that is not a valid URI, and `URI.create` throws on it. On Spark 4.0+ that exception escaped `CometWriteFiles.convert` and failed the query. On 3.x `CometDataWritingCommand.convert` caught it and silently handed the write back to Spark, so the native writer was never used for those paths. Round-trip through `Path` instead, which re-escapes. Only the scheme and authority reach `extractObjectStoreOptions`, but parsing has to succeed to get at them. Also assert that the INSERT INTO visibility test's write actually went native, rather than inferring it from the read-back.
The raw/decoded URI comparison only catches what java.net.URI had to
escape, and java.net.URI leaves non-ASCII path characters alone, so an
hdfs://ns/cafe<U+0301>/output destination was admitted. percent_encoding's
should_percent_encode is !byte.is_ascii() || set.contains(byte), so the
native parser escapes every non-ASCII byte regardless of the encode set
and the writer creates caf%C3%A9 outside Spark's staging directory.
The guard now also declines any character the native parser rewrites. The
ASCII half of that set was determined against the locked url 2.5 crate by
parsing hdfs://ns/pre<c>post/output for every printable ASCII c: space, ",
#, <, >, ?, backtick, { and } are rewritten and the rest survive, so
partition directories and Spark's _temporary attempt paths still qualify.
The comment no longer claims the Java comparison detects the divergence on
its own; both conditions are kept because the Java one still catches a
literal % that the native parser leaves alone.
Tests add accented (precomposed and combining), CJK, emoji and nested
non-ASCII cases plus the remaining escaped ASCII characters, built from
code points since scalastyle forbids non-ASCII source. Disabling the new
condition makes the accented case fail, so the Java comparison alone
demonstrably does not cover it.
github-actions
Bot
added
enhancement
Correctness:
- Decline HDFS writes whose *file names* would diverge, not just their
directory. `mapreduce.output.basename` is caller-controlled and reaches every
committed name through `HadoopMapReduceCommitProtocol.getFilename`; a basename
holding `?` or `#` makes the native URL parser truncate, so every task writes
the same name and they overwrite each other at commit. Adds an execution-time
backstop over the complete `newTaskTempFile` path, which a custom commit
protocol owns and planning cannot predict.
- Read the compression option case-insensitively, as Spark's `ParquetOptions`
does. `option("Compression", "lz4_raw")` used to fall through to the SQLConf
default, so the unsupported-codec guard was bypassed and Comet wrote SNAPPY
into a file Spark had named `.lz4raw.parquet`. The codec is now also
re-derived per task from `CodecConfig.from(taskAttemptContext)`, the same
place the file extension comes from, so the name and the contents agree by
construction. The shared helpers move to `NativeWriteUtils`, which fixes the
identical bug on the Spark 3.x path.
- Use `Utils.tryWithSafeFinallyAndFailureCallbacks` / `tryWithSafeFinally` in
`executeTask` and `writeNatively`, matching `FileFormatWriter`: a failure
while aborting or closing the iterator is attached as a suppressed exception
instead of replacing the failure that caused it. `statsTrackers` moves inside
the guard so a throwing `newTaskInstance` still reaches `abortTask`.
Planning and reporting:
- Convert `WriteFilesExec` from its enclosing `DataWritingCommandExec` rather
than from a separate tag pre-pass, so the output path comes straight from the
command that owns it and neither `withNewChildren` copying tags nor "nothing
hands us a bare WriteFilesExec" has to hold.
- Only skip the fallback reason on `DataWritingCommandExec` when its child
really was converted; a write with no native child now says why it fell back.
- Drop the node's duplicate `files_written`/`bytes_written`/`rows_written`.
`BasicWriteJobStatsTracker` is authoritative here, and the native
`bytes_written` reads 0 on HDFS.
Tests and docs:
- Rust `url_path_rewritten_characters` pins the `url` crate's path encode set,
which the JVM guard mirrors; a crate upgrade can no longer reopen the hole
with a green build.
- New JVM coverage: mixed-case `compression` (honored, and declined when
unsupported), the apache#3426 nested-name INSERT, an empty non-zero partition
writing no file, the third-party `WriteTaskStatsTracker` warning, and the
basename/committer-path guards.
- The abort test no longer claims to demonstrate task retry or speculation.
- `installation.md` says which Spark versions its EXPLAIN output applies to.
915b174 to
278f28e
Compare
peterxcli
changed the title
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
#5763 landed on the 19th as 5442c937d, so this branch still carries its pre-merge commits and conflicts with main in six files. Replaying only 278f28e0a and e9f5e5e7f onto main conflicts just in the CometDataWritingCommand imports, where #5821 added the hasEmptyRelationInput guard, and in the two suite lists. Could you rebase down to those two commits? Two comments describe the old 3.x writer and stop being true with this change. NativeWriteUtils.escapedHdfsDestination says that on 3.x Comet names the files itself and the basename is always part, and the ParquetWriterExec field docs say work_dir is the Spark 3.x path. After this nothing sets work_dir, job_id or task_attempt_id, so could the Some(work_dir) arm in ParquetWriterExec::execute go as well, with the three proto fields reserved?
The task closure calls createTaskContext, which is an instance method because it reads jobTrackerID. That pulls the whole CometNativeWriteExec, child subtree included, into every task despite the captured* locals, which is the thing CometWriteFilesExec goes out of its way to avoid. Could createTaskContext move to the companion object and take jobTrackerID as a parameter?
runNativeWriteJob now runs the same task loop as CometWriteFilesExec, but it leaves out the two empty-input cases that one copies from FileFormatWriter. A zero-partition child runs no task at all, so the output gets a _SUCCESS and no schema-bearing file, and spark.read.parquet on it fails to infer a schema (SPARK-23271, #5303). And every empty partition still writes its own file, because create_arrow_writer runs before the first batch. Could this take the same parallelize(Seq.empty[ColumnarBatch], 1) swap and the sparkPartitionId == 0 || batches.hasNext check, and drop the assume(isSpark40Plus) from the SPARK-23271 and empty-partition tests in CometParquetWriterSuite?
apache#5763 landed as 5442c93, so the branch's pre-merge copies of its commits are superseded by main's squashed version. The tree matches replaying only 278f28e and e9f5e5e onto main, with the CometDataWritingCommand imports and the two suite lists resolved.
Both native writers now take the exact file name from Spark's commit protocol, so nothing sets work_dir, job_id or task_attempt_id any more. Remove the Some(work_dir) arm from ParquetWriterExec::execute, reserve the three proto fields, and correct the comments that still described the Spark 3.x writer naming its own files.
Mirror CometWriteFilesExec and FileFormatWriter: swap a zero-partition child for a single empty partition so the output still carries a schema (SPARK-23271), and only write a file from partition 0 or a partition that has rows. Move createTaskContext to the companion object so the task closure no longer captures the exec node and its child subtree. The SPARK-23271 and empty-partition tests now run on Spark 3.x too.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Summary
- Prior state and problem: The legacy writer bypassed Spark’s configured commit protocol, discarded its chosen filename, and handled job completion inconsistently between execution entry points.
- Design approach: Spark 3.x now runs the configured job/task lifecycle. The included Spark 4 prerequisite replaces
WriteFilesExec, leaving job commit, SaveMode handling and catalog refresh with Spark. - Correctness / compatibility analysis: Compared the write contracts against Spark 3.4.3, 3.5.9, 4.0.4, 4.1.3 and 4.2.0. Filename ownership, task identifiers, commit-message delivery and exception preservation follow Spark’s contracts. The existing P2 concerning percent-bearing HDFS basenames remains in this checkout. A local probe using the current guard methods and real Hadoop
Pathconfirmed thatpart%fooandpart%25pass planning but fail the task guard. Ordinarypartsucceeds. This duplicates an existing finding, so no new inline finding is returned. - Key design decisions: Separate version paths are justified by Spark’s concrete-class versus base-trait write discovery. Explicit task state keeps the Spark 4 serialization boundary clear. The existing Spark 3 closure-capture concern remains visible. No measured performance regression was established, and the per-row statistics callback cost remains unmeasured.
- Implementation sketch: Prepare the Hadoop job, allocate the committer’s exact filename, execute and close the native iterator, then commit the task and deliver its message through
runJob. Failures preserve the original exception while invoking abort callbacks. - Behavioral changes worth calling out: Both Spark 3 execution entry points now complete job commit, and filenames follow Spark’s convention. Spark 4 retains Spark’s surrounding write framework. Existing Spark 3 empty-input concerns are already covered by the prior review.
- Suggested improvements: Resolve the existing HDFS basename P2 by applying the Java URI-escaping check to the basename during planning, allowing these writes to fall back before execution.
Reviewed the entire 21-file diff from 481aefea9c60592612650de73bd2e1c7aa173979 to e9f5e5e7fe22baa4d5ff74499a13c7c72aa3beb6, including prerequisites. Confirmed non-draft status and read existing discussion. Routed skill: review-comet-pr. No sibling skill applies to this operator review.
Exact-head CI: 54 successful checks and 10 skipped, with no failures. The inspected CI merge e40331c50bd2f0435798837b835249ec019596ac has the same full tree as the reviewed head. Logs confirm all eight Spark 3.4 commit-lifecycle tests passed, plus 39 writer tests with nine cancellations. Spark 4.0 passed all 48 writer tests.
Local validation: the six-case path probe, suite inventory check and git diff --check passed. No full native/JVM build was run locally. System Maven rejected the repository’s maven.config. No local HDFS integration, automatic retry, speculation or throughput benchmark was run.
No additional introduced P1/P2 issues found within this review. Existing blockers remain.
|
Thanks, all addressed:
With the swap in place, the |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Summary
- Prior state and problem: The Spark 3.x writer hardcoded the committer, discarded its allocated filename and handled job completion inconsistently between execution entry points.
- Design approach: Use Spark’s configured
FileCommitProtocol, prepared Hadoop configuration and exact task filename. Return task commit messages throughrunJob. - Correctness / compatibility analysis: Compared lifecycle and filename contracts against Spark 3.4.3 and 3.5.9, and checked the shared native path against supported Spark 4.x sources. Task identifiers, message delivery and cleanup ordering follow those contracts. The earlier HDFS basename, closure-capture and empty-input concerns are addressed in the current tree.
- Key design decisions: Moving task-context creation to the companion object avoids capturing the execution plan. Removing the accumulator, iterator wrapper and native filename-generation branch simplifies ownership. No introduced performance regression was established. Throughput was not benchmarked.
- Implementation sketch: Prepare the job, allocate the committer’s filename, execute and close the native iterator, then commit the task and job. Failure paths abort while preserving the original exception.
- Behavioral changes worth calling out: Both execution entry points complete job commit. Zero-partition input gets a schema-bearing file, and empty nonzero partitions skip file creation. Removed protobuf field numbers and names are reserved.
- Suggested improvements: None at P1/P2 severity.
Reviewed the entire 10-file diff from a86c9672a63c909f0fd7b752c86b5a679df4e84b to 086bc1eefc49dfa694e210c077984533e8de8450. Confirmed non-draft status and read the existing reviews and discussion. Routed skills: review-comet-pr, review-comet-expression-pr, review-comet-ffi-pr and review-comet-memory-pr.
Exact-head CI: Checks attached to this SHA currently show 20 successful, 13 skipped and two running, with no failures. The running checks are Spark 4.1 execution and expression suites. Inspected logs confirm 49 Parquet writer tests, the empty-relation writer test and four native writer tests passed. All eight Spark 3.x lifecycle tests were canceled by their version gate. CI actually checked out merge 110c2dc6711a86ac089d710eddf9eb23041c19a5, which includes four additional changed files from main.
Validation: An eight-case probe compiled from the current path-guard methods passed using real Hadoop Path, including both previously failing percent-bearing basenames. Suite registration and git diff --check passed. No full local native/JVM build was run. Earlier Spark 3.x runtime evidence predates the latest edits. Spark SQL CI was skipped, and real HDFS, retries, speculation and performance were not validated locally.
No introduced P1/P2 issues found within this review. No substantiated existing P1/P2 blocker remains.
4 participants
Which issue does this PR close?
Addresses the remaining Spark 3.x commit-protocol gaps in #2827 and #3015. This does not claim to close their remaining version and feature gaps.
Rationale for this change
Spark 3.4/3.5 native writes replace the whole
DataWritingCommandExec. The retained writer hardcodes the commit protocol, discards the filename returned bynewTaskTempFile, and does not consistently complete or abort the job from both execution entry points. Committers that track individual filenames can therefore commit a different file from the one native code writes.Dependency: #5763 is still open. The writer implementation is based on its current head,
5f5397f5049b1bfc6fe27e6c8e926ea2d0bee959, and the PR remains draft. The branch also includes the existing merge frommainat481aefea9. Functional follow-up changes are limited to the Spark 3.x writer, with focused tests and their CI registrations. Compare the writer implementation against the dependency. Once #5763 merges, this follow-up can be rebased onto main.What changes are included in this PR?
fileFormat.prepareWrite, and instantiate Spark's configuredFileCommitProtocol. Task contexts inherit the prepared configuration, including changes fromsetupJob.newTaskTempFilefilename through the existingoutput_pathfield withwork_dirunset. The previoustask_output_pathproposal is removed; protobuf field 9 remainsoutput_schema. This follow-up changes no protobuf or native Rust code.runJob, invokeonTaskCommit, and remove the collection accumulator.CometWriteFilesExecpath.CometNativeWriteSuitebeside the Parquet writer suite in both Linux and macOS CI matrices. This fixes the Preflight missing-suite failure.How are these changes tested?
The runtime results below were collected on writer commit
278f28e0a, before the subsequent merge frommainand CI registration fix. Built the native library withmake corebefore JVM tests. Maven ran from the repository root, without-pl, using JDK 17 and a debug native build. Spark profiles were built clean when switching versions.CometNativeWriteSuiteCometParquetWriterSuiteThe new suite uses a configured committer with an unusual exact filename containing spaces and
%, checks job-option/file-format preparation and task-message delivery, and verifies native row metrics and output read-back for both entry points. It injects failures in native iterator construction, native batch execution, final native metrics cleanup,commitTask,onTaskCommit, andcommitJob, including throwing abort callbacks. Assertions check task/job abort, preservation of the original exception, suppressed cleanup errors,_SUCCESS, and staging cleanup. Existing success-marker, Spark filename, and task-abort checks now also run on Spark 3.x; the writer suite covers SaveMode, schema/field IDs, codecs, and local URI regressions.Spotless, Scalastyle, and
git diff --checkpassed.After the merge from
main, CI registration commite9f5e5e7fpassed the suite inventory check, CI configuration checks, all 15 Iceberg shard validation tests, benchmark runner checks, andactionlint --shellcheck=off. A clean Spark 3.5/Scala 2.12/JDK 17 package build and semantic Scalafix check also passed (-DskipTests; runtime tests were not repeated for the two workflow-only additions).Limitations: validation used local Spark execution, not a real HDFS cluster, automatic task retries, or speculative attempts. #5763's unresolved percent-bearing HDFS basename admission issue remains a dependency limitation. This does not extend the legacy writer's catalog refresh, partitioning, file rolling, or other experimental writer behavior.